package org.lk.common.factory.impl;

import cn.hutool.core.util.StrUtil;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.lk.common.reader.CustomDebeziumDeserializationSchema;
import org.lk.common.factory.StreamReader;
import org.lk.entity.DataSourceEntity;
import org.lk.common.wrapper.Wrapper;

import java.util.ArrayList;
import java.util.List;

/**
 * @author: lk
 * @since: mysql流式读取
 */
public class MysqlStreamReader extends StreamReader {


    public MysqlStreamReader(DataSourceEntity dataSourceEntity) {
        super(dataSourceEntity);
        wrapper = new Wrapper();
    }

    @Override
    public String wrapperTableName(String tableName) {
        return wrapper.wrap(dataSourceEntity.getDsSchema() + "." + tableName);
    }

    @Override
    public DebeziumSourceFunction<String> readerSourceCdc(List<String> tableNameList, StartupOptions startupOptions) {
        String allTable = tableNameList.get(0);
        String[] wrapperTableList;
        if (StrUtil.isNotEmpty(allTable) && allTable.equalsIgnoreCase("*")) {
            wrapperTableList = new String[]{allTable};
        } else {
            List<String> wrapperTables = new ArrayList<>();
            for (String tableName : tableNameList) {
                wrapperTables.add(wrapperTableName(tableName));
            }
            wrapperTableList = wrapperTables.toArray(new String[]{});
        }
        return MySqlSource.<String>builder()
                .hostname(dataSourceEntity.getDsHost())
                .port(dataSourceEntity.getDsPort())
                .databaseList(dataSourceEntity.getDsSchema())
                .tableList(wrapperTableList)
                .username(dataSourceEntity.getUsername())
                .password(dataSourceEntity.getPassword())
                .deserializer(new CustomDebeziumDeserializationSchema())
                .startupOptions(startupOptions).build();
    }
}
